package cn.itcast.flink.base;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.metrics.Counter;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.util.Collector;

import java.text.SimpleDateFormat;
import java.util.Iterator;
import java.util.Map;
import java.util.Random;
import java.util.UUID;

/**
 * Author itcast
 * Date 2021/8/2 15:06
 * Desc 自动好评定时触发任务的案例
 */
public class OrderAutomationFavorable {
    public static void main(String[] args) throws Exception {
        //ParameterTool.fromArgs()
        //1.创建流执行环境，设置并行度
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //开启 blink streaming 运行环境
        env.setRuntimeMode(RuntimeExecutionMode.STREAMING);
        //开启 checkpoint
        env.enableCheckpointing(1000);
        env.setStateBackend(new FsStateBackend("file:///d:/chk"));
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup
                .RETAIN_ON_CANCELLATION);
        env.getCheckpointConfig().setCheckpointInterval(500);
        env.getCheckpointConfig().setCheckpointTimeout(60000);
        env.getCheckpointConfig().setTolerableCheckpointFailureNumber(10);
        //开启重启策略
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(10, 5000L));
        //2.创建数据源
        DataStreamSource<Tuple3<String, String, Long>> source = env.addSource(new MySource());
        //定义一个触发时间，如果超过 5 秒中，将当前的订单的评论置为默认好评
        long interval = 5L;
        //3.transformation
        //todo 设置假如 interval=5秒，若用户未对订单做出评价，自动给与好评. Tuple3<用户id, 订单id, 订单生成时间>
        SingleOutputStreamOperator<Object> result = source.keyBy(t -> t.f0)
                //根据 用户id 进行分组，然后处理 TimerProcessFunction:KeyedProcessFunction
                .process(new TimerProcessFunction(interval));
        //4.执行
        env.execute();
    }

    /**
     * 自定义source实时产生订单数据Tuple3<用户id,订单id,订单生成时间>
     */
    public static class MySource implements SourceFunction<Tuple3<String, String, Long>> {
        private boolean flag = true;

        @Override
        public void run(SourceContext<Tuple3<String, String, Long>> ctx) throws Exception {
            Random random = new Random();
            while (flag) {
                String userId = random.nextInt(5) + "";
                String orderId = UUID.randomUUID().toString();
                long currentTimeMillis = System.currentTimeMillis();
                ctx.collect(Tuple3.of(userId, orderId, currentTimeMillis));
                Thread.sleep(500);
            }
        }

        @Override
        public void cancel() {
            flag = false;
        }
    }

    /**
     * 自定义处理函数用来给超时订单做自动好评!
     * 如一个订单进来:<订单id, 2020-10-10 12:00:00>
     * 那么该订单应该在12:00:00 + 5s 的时候超时!
     * 所以我们可以在订单进来的时候设置一个定时器,在订单时间 + interval的时候触发!
     * KeyedProcessFunction<K, I, O>
     * KeyedProcessFunction<String, Tuple3<用户id, 订单id, 订单生成时间>, Object>
     */
    private static class TimerProcessFunction extends KeyedProcessFunction<String, Tuple3<String, String, Long>, Object> {
        MapState<String, Long> mapState = null;
        //创建构造器
        private Long interval = 0L;

        public TimerProcessFunction(Long _interval) {
            this.interval = _interval;
        }

        @Override
        public void open(Configuration parameters) throws Exception {

            MapStateDescriptor<String, Long> orderStateDesc = new MapStateDescriptor<>(
                    "orderstate",
                    String.class,
                    Long.class
            );
            //获取当前上下文中 state
            mapState = getRuntimeContext().getMapState(orderStateDesc);
        }

        @Override
        public void processElement(Tuple3<String, String, Long> value, Context ctx, Collector<Object> out) throws Exception {
            mapState.put(value.f1, value.f2);
            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            System.out.println("处理的时间value:"+sdf.format(value.f2));
            //注册触发器
            ctx.timerService().registerProcessingTimeTimer(value.f2 + interval);
        }

        // 超过了注册器的时间，才会执行 onTimer 方法
        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<Object> out) throws Exception {
            //能够执行到这里说明订单超时了!超时了得去看看订单是否评价了(实际中应该要调用外部接口/方法查订单系统!,
            Iterator<Map.Entry<String, Long>> iterator = mapState.iterator();
            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            System.out.println("执行的窗口时间："+sdf.format(timestamp));
            Long ctxTime = ctx.timestamp();
            System.out.println("触发器计算时间："+sdf.format(ctxTime));
            while (iterator.hasNext()) {

                //获取当前的超时的订单
                Map.Entry<String, Long> entry = iterator.next();
                //获取每个超时的订单id
                String orderId = entry.getKey();
                //调用外部接口，这个接口实现 传入一个 orderId 将当前订单的评价改为默认好评
                boolean isEvaluate = isEvaluate(orderId);
                //在生产环境下，可以去查询相关的订单系统.
                if (isEvaluate) {
                    System.out.println("当前订单：" + orderId + " 默认五星好评！");
                    //mapState.remove(currentKey);
                }
            }
            //我们这里没有接口调用，直接模拟查询订单
            //没有评价才给默认好评!并直接输出提示! 已经评价了,直接输出提示!
        }

        /**
         * 传入一个 orderId 返回一个是否改成功的值
         *
         * @param orderId
         * @return
         */
        private boolean isEvaluate(String orderId) {
            return orderId.hashCode() % 3 == 0;
        }
    }
}
